package com.qezhhnjy.demo.hadoop.service;


import com.alibaba.fastjson.JSON;
import com.qezhhnjy.demo.hadoop.config.HadoopConfig;
import com.qezhhnjy.demo.hadoop.pool.HadoopConn;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author zhaoyangfu - 2021/2/3 17:37
 */
@Slf4j
@Component
public class HadoopService {

    @Resource
    private HadoopConn   conn;
    @Resource
    private HadoopConfig config;

    /**
     * 创建Hdfs目录
     *
     * @param path Hdfs的相对目录路径，比如：/testDir
     * @return boolean 是否创建成功
     * @author adminstrator
     * @since 1.0.0
     */
    public boolean mkdir(String path) {
        //如果目录已经存在，则直接返回
        if (checkExists(path)) {
            return true;
        } else {
            try {
                FileSystem fileSystem = conn.open();
                //最终的Hdfs文件目录
                String hdfsPath = generateHdfsPath(path);
                //创建目录
                return fileSystem.mkdirs(new Path(hdfsPath));
            } catch (Exception e) {
                log.error("创建Hdfs目录失败，path:{}", path, e);
                return false;
            } finally {
                conn.close();
            }
        }
    }

    /**
     * 上传文件至Hdfs
     *
     * @param srcFile 本地文件路径，比如：D:/test.txt
     * @param dstPath Hdfs的相对目录路径，比如：/testDir
     * @author adminstrator
     * @since 1.0.0
     */
    public void uploadFileToHdfs(String srcFile, String dstPath) {
        this.uploadFileToHdfs(false, true, srcFile, dstPath);
    }

    /**
     * 上传文件至Hdfs
     *
     * @param delSrc    是否删除本地文件
     * @param overwrite 是否覆盖Hdfs上面的文件
     * @param srcFile   本地文件路径，比如：D:/test.txt
     * @param dstPath   Hdfs的相对目录路径，比如：/testDir
     * @author adminstrator
     * @since 1.0.0
     */
    public void uploadFileToHdfs(boolean delSrc, boolean overwrite, String srcFile, String dstPath) {
        //源文件路径
        Path localSrcPath = new Path(srcFile);
        //目标文件路径
        Path hdfsDstPath = new Path(generateHdfsPath(dstPath));

        try {
            FileSystem fileSystem = conn.open();
            fileSystem.copyFromLocalFile(delSrc, overwrite, localSrcPath, hdfsDstPath);
        } catch (Exception e) {
            log.error("上传文件至Hdfs失败，srcFile:{},dstPath:{}", srcFile, dstPath, e);
        } finally {
            conn.close();
        }
    }

    /**
     * 判断文件或者目录是否在Hdfs上面存在
     *
     * @param path Hdfs的相对目录路径，比如：/testDir、/testDir/a.txt
     * @return boolean
     * @author adminstrator
     * @since 1.0.0
     */
    public boolean checkExists(String path) {
        try {
            FileSystem fileSystem = conn.open();

            //最终的Hdfs文件目录
            String hdfsPath = generateHdfsPath(path);

            //创建目录
            return fileSystem.exists(new Path(hdfsPath));
        } catch (Exception e) {
            log.error("'判断文件或者目录是否在Hdfs上面存在'失败，path:{}", path, e);
            return false;
        } finally {
            conn.close();
        }
    }

    /**
     * 获取Hdfs上面的某个路径下面的所有文件或目录（不包含子目录）信息
     *
     * @param path Hdfs的相对目录路径，比如：/testDir
     * @return java.util.List<java.util.Map < java.lang.String, java.lang.Object>>
     * @author adminstrator
     * @since 1.0.0
     */
    public List<Map<String, Object>> listFiles(String path, PathFilter pathFilter) {
        //返回数据
        List<Map<String, Object>> result = new ArrayList<>();

        //如果目录已经存在，则继续操作
        if (checkExists(path)) {

            try {
                FileSystem fileSystem = conn.open();

                //最终的Hdfs文件目录
                String hdfsPath = generateHdfsPath(path);

                FileStatus[] statuses;
                //根据Path过滤器查询
                if (pathFilter != null) {
                    statuses = fileSystem.listStatus(new Path(hdfsPath), pathFilter);
                } else {
                    statuses = fileSystem.listStatus(new Path(hdfsPath));
                }

                if (statuses != null) {
                    for (FileStatus status : statuses) {
                        //每个文件的属性
                        Map<String, Object> fileMap = new HashMap<>(2);

                        fileMap.put("path", status.getPath().toString());
                        fileMap.put("isDir", status.isDirectory());

                        result.add(fileMap);
                    }
                }
            } catch (Exception e) {
                log.error("获取Hdfs上面的某个路径下面的所有文件失败，path:{}", path, e);
            } finally {
                conn.close();
            }
        }

        return result;
    }


    /**
     * 从Hdfs下载文件至本地
     *
     * @param srcFile Hdfs的相对目录路径，比如：/testDir/a.txt
     * @param dstFile 下载之后本地文件路径（如果本地文件目录不存在，则会自动创建），比如：D:/test.txt
     * @author adminstrator
     * @since 1.0.0
     */
    public void downloadFileFromHdfs(String srcFile, String dstFile) {
        //Hdfs文件路径
        Path hdfsSrcPath = new Path(generateHdfsPath(srcFile));
        //下载之后本地文件路径
        Path localDstPath = new Path(dstFile);

        try {
            FileSystem fileSystem = conn.open();

            fileSystem.copyToLocalFile(hdfsSrcPath, localDstPath);
        } catch (Exception e) {
            log.error("从Hdfs下载文件至本地失败，srcFile:{},dstFile:{}", srcFile, dstFile, e);
        } finally {
            conn.close();
        }
    }

    /**
     * 打开Hdfs上面的文件并返回 InputStream
     *
     * @param path Hdfs的相对目录路径，比如：/testDir/c.txt
     * @return FSDataInputStream
     * @author adminstrator
     * @since 1.0.0
     */
    public FSDataInputStream open(String path) {
        //Hdfs文件路径
        Path hdfsPath = new Path(generateHdfsPath(path));

        try {
            FileSystem fileSystem = conn.open();

            return fileSystem.open(hdfsPath);
        } catch (Exception e) {
            log.error("打开Hdfs上面的文件失败，path:{}", path, e);
        } finally {
            conn.close();
        }

        return null;
    }

    /**
     * 打开Hdfs上面的文件并返回byte数组，方便Web端下载文件
     * <p>new ResponseEntity<byte[]>(byte数组, headers, HttpStatus.CREATED);</p>
     * <p>或者：new ResponseEntity<byte[]>(FileUtils.readFileToByteArray(templateFile), headers, HttpStatus.CREATED);</p>
     *
     * @param path Hdfs的相对目录路径，比如：/testDir/b.txt
     * @return FSDataInputStream
     * @author adminstrator
     * @since 1.0.0
     */
    public byte[] openWithBytes(String path) throws IOException {
        //Hdfs文件路径
        Path hdfsPath = new Path(generateHdfsPath(path));
        FSDataInputStream input = null;
        try {
            FileSystem fileSystem = conn.open();
            input = fileSystem.open(hdfsPath);
            return IOUtils.toByteArray(input);
        } catch (Exception e) {
            log.error("打开Hdfs上面的文件失败，path:{}", path, e);
        } finally {
            conn.close();
            if (input != null) input.close();
        }

        return null;
    }

    /**
     * 打开Hdfs上面的文件并返回String字符串
     *
     * @param path Hdfs的相对目录路径，比如：/testDir/b.txt
     * @return FSDataInputStream
     * @author adminstrator
     * @since 1.0.0
     */
    public String openWithString(String path) throws IOException {
        //Hdfs文件路径
        Path hdfsPath = new Path(generateHdfsPath(path));

        FSDataInputStream inputStream = null;
        try {
            FileSystem fileSystem = conn.open();
            inputStream = fileSystem.open(hdfsPath);

            return IOUtils.toString(inputStream, StandardCharsets.UTF_8);
        } catch (Exception e) {
            log.error("打开Hdfs上面的文件失败，path:{0}", path, e);
        } finally {
            conn.close();
            if (inputStream != null) {
                inputStream.close();
            }
        }

        return null;
    }

    /**
     * 打开Hdfs上面的文件并转换为Java对象（需要Hdfs上门的文件内容为JSON字符串）
     *
     * @param path Hdfs的相对目录路径，比如：/testDir/c.txt
     * @return FSDataInputStream
     * @author adminstrator
     * @since 1.0.0
     */
    public <T> T openWithObject(String path, Class<T> clazz) throws IOException {
        //1、获得文件的json字符串
        String jsonStr = this.openWithString(path);
        return JSON.parseObject(jsonStr, clazz);
    }

    /**
     * 重命名
     *
     * @param srcFile 重命名之前的Hdfs的相对目录路径，比如：/testDir/b.txt
     * @param dstFile 重命名之后的Hdfs的相对目录路径，比如：/testDir/b_new.txt
     * @author adminstrator
     * @since 1.0.0
     */
    public boolean rename(String srcFile, String dstFile) {
        //Hdfs文件路径
        Path srcFilePath = new Path(generateHdfsPath(srcFile));
        //下载之后本地文件路径
        Path dstFilePath = new Path(dstFile);

        try {
            FileSystem fileSystem = conn.open();
            return fileSystem.rename(srcFilePath, dstFilePath);
        } catch (Exception e) {
            log.error("重命名失败，srcFile:{},dstFile:{}", srcFile, dstFile, e);
        } finally {
            conn.close();
        }

        return false;
    }

    /**
     * 删除Hdfs文件或目录
     *
     * @param path Hdfs的相对目录路径，比如：/testDir/c.txt
     * @return boolean
     * @author adminstrator
     * @since 1.0.0
     */
    public boolean delete(String path) {
        //Hdfs文件路径
        Path hdfsPath = new Path(generateHdfsPath(path));

        try {
            FileSystem fileSystem = conn.open();
            return fileSystem.delete(hdfsPath, true);
        } catch (Exception e) {
            log.error("删除Hdfs文件或目录失败，path:{}", path, e);
        } finally {
            conn.close();
        }

        return false;
    }

    /**
     * 获取某个文件在Hdfs集群的位置
     *
     * @param path Hdfs的相对目录路径，比如：/testDir/a.txt
     * @return org.apache.hadoop.fs.BlockLocation[]
     * @author adminstrator
     * @since 1.0.0
     */
    public BlockLocation[] getFileBlockLocations(String path) {
        //Hdfs文件路径
        Path hdfsPath = new Path(generateHdfsPath(path));

        try {
            FileSystem fileSystem = conn.open();
            FileStatus fileStatus = fileSystem.getFileStatus(hdfsPath);
            return fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
        } catch (Exception e) {
            log.error("获取某个文件在Hdfs集群的位置失败，path:{}", path, e);
        } finally {
            conn.close();
        }

        return null;
    }


    /**
     * 将相对路径转化为Hdfs文件路径
     *
     * @param dstPath 相对路径，比如：/data
     * @return java.lang.String
     * @author adminstrator
     * @since 1.0.0
     */
    private String generateHdfsPath(String dstPath) {
        String hdfsPath = config.uri();
        if (dstPath.startsWith("/")) {
            hdfsPath += dstPath;
        } else {
            hdfsPath = hdfsPath + "/" + dstPath;
        }
        return hdfsPath;
    }
}